package com.rxjava;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

/**
 * Created by kkmike999 on 16/5/18.
 */
public class merge {
	public static void main(String[] args) throws InterruptedException {
		Observable o0 = Observable.create(new Observable.OnSubscribe<String>() {
			@Override
			public void call(Subscriber<? super String> subscriber) {
				System.out.println("o0:" + Thread.currentThread() + "  " + System.currentTimeMillis());

//				try {
//					Thread.sleep(80);
//				} catch (InterruptedException e) {
//					e.printStackTrace();
//				}

				for (int i = 0; i < Integer.MAX_VALUE; i++) {

				}

				System.out.println("for结束" + "  " + System.currentTimeMillis());

				subscriber.onNext("o0");
			}
		})
		                          .subscribeOn(Schedulers.newThread());

		Observable o1 = Observable.create(new Observable.OnSubscribe<String>() {
			@Override
			public void call(Subscriber<? super String> subscriber) {
				System.out.println("o1:" + Thread.currentThread() + "  " + System.currentTimeMillis());
				subscriber.onError(new RuntimeException("错误" + "  " + System.currentTimeMillis()));
			}
		})
		                          .subscribeOn(Schedulers.newThread());

		Observable.merge(o0, o1)
//		          .subscribeOn(Schedulers.io())
                  .observeOn(Schedulers.newThread())
                  .subscribe(new Subscriber<String>() {
	                  @Override
	                  public void onCompleted() {}

	                  @Override
	                  public void onError(Throwable throwable) {
		                  System.out.println("onError:" + Thread.currentThread() + "  " + System.currentTimeMillis());
		                  System.out.println(throwable.getMessage());
	                  }

	                  @Override
	                  public void onNext(String s) {
		                  System.out.println("onNext:" + Thread.currentThread() + "  " + System.currentTimeMillis());
		                  System.out.println(s);

		                  try {
			                  Thread.sleep(80);
		                  } catch (InterruptedException e) {
			                  e.printStackTrace();
		                  }
	                  }
                  });

		Thread.sleep(200);
	}
}
